package zjs.dc.controller;

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

/**
 * @Class:
 * @Author: kongcb
 * @Description: TODO
 * @Date: 2021/12/15 9:23
 */
public class TTender {
    public static void main(String[] args) {
        SparkSession spark = SparkSession.builder().appName("t_tender").enableHiveSupport().getOrCreate();
        spark.sparkContext().setLogLevel("ERROR");
        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_invite_info");
//        Dataset<Row> zs_data = spark.sql("select * from a_dc_ep_ods.zs_t_bidding_baseinfo");
        long zs_count = zs_data.count();
        if(zs_count>0){
//            //企查查招投标数据表取统一社会信用代码
//            spark.sql("select a.* ,b.credit_code from a_dc_ep_dwi.t_tender a left join a_dc_ep_ods.t_eci_company b on a.company_id = b.company_id where b.company_id is not null")
//                    .createOrReplaceTempView("qcc_tender");
//
//            //中数招投标信息去统一社会信用代码
//            spark.sql("select a.record_id,a.jobid,a.creditcode,b.title,b.provinces,b.notice_date from a_dc_ep_ods.zs_t_bidding_party a left join a_dc_ep_ods.zs_t_bidding_baseinfo b on a.docid = b.docid where b.record_id is not null and a.creditcode !='' and a.creditcode is not null")
//                    .createOrReplaceTempView("zs_tender");
//
//            spark.sql("select aaa.key_no,aaa.company_id,aaa.company_name,aaa.credit_code from " +
//                    "(select aa.key_no,aa.company_id,aa.company_name,aa.credit_code,ROW_NUMBER() OVER(PARTITION BY aa.credit_code ORDER BY aa.dates DESC,aa.isadd DESC) num " +
//                    "from a_dc_ep_ods.t_eci_company aa where aa.credit_code is not null) aaa where aaa.num = 1")
//                    .createOrReplaceTempView("tmp_t_eci_company");
//
//            //更新
//            spark.sql("insert overwrite table a_dc_ep_incr.t_tender select " +
//                    "zs.record_id," +
//                    "qcc.key_no," +
//                    "qcc.company_id," +
//                    "qcc.company_name," +
//                    "zs.title," +
//                    "zs.notice_date," +
//                    "zs.provinces," +
//                    "''," +
//                    "zs.jobid," +
//                    "'1' as isadd " +
//                    "from zs_tender zs left join " +
//                    "tmp_t_eci_company qcc " +
//                    "on upper(trim(qcc.credit_code)) = upper(trim(zs.creditcode)) where qcc.credit_code is not null");
//            //删除
//            spark.sql("insert into table a_dc_ep_incr.t_tender select " +
//                    "qcc.id,\n" +
//                    "qcc.key_no,\n" +
//                    "qcc.company_id,\n" +
//                    "qcc.company_name,\n" +
//                    "qcc.title,\n" +
//                    "qcc.public_date,\n" +
//                    "qcc.province_name,\n" +
//                    "qcc.channel_name,\n" +
//                    "qcc.dates," +
//                    "'-1' as isadd " +
//                    "from qcc_tender qcc where qcc.id in " +
//                    "(select a.record_id from a_dc_ep_ods.zs_t_bidding_baseinfo_del a left join a_dc_ep_ods.zs_t_bidding_baseinfo b " +
//                    "on a.record_id = b.record_id where b.record_id is null)");

            //企查查招投标数据表取统一社会信用代码
            spark.sql("select a.* ,b.credit_code from a_dc_ep_dwi.t_tender a left join a_dc_ep_ods.t_eci_company b on a.company_id = b.company_id where b.company_id is not null")
                    .createOrReplaceTempView("qcc_tender");

            //中数招投标信息去统一社会信用代码
            spark.sql("select a.*,b.creditcode from a_dc_ep_ods.zs_t_invite_info a left join a_dc_ep_ods.zs_t_invite_info_del b on a.entid = b.entid and a.INVITE_PUB_DATETIME=b.INVITE_PUB_DATETIME where b.entid is not null")
                    .createOrReplaceTempView("zs_tender");

            spark.sql("select aaa.key_no,aaa.company_id,aaa.company_name,aaa.credit_code from " +
                            "(select aa.key_no,aa.company_id,aa.company_name,aa.credit_code,ROW_NUMBER() OVER(PARTITION BY aa.credit_code ORDER BY aa.dates DESC,aa.isadd DESC) num " +
                            "from a_dc_ep_ods.t_eci_company aa where aa.credit_code is not null) aaa where aaa.num = 1")
                    .createOrReplaceTempView("tmp_t_eci_company");

            //更新
            spark.sql("insert overwrite table a_dc_ep_incr.t_tender select " +
                    "zs.record_id," +
                    "qcc.key_no," +
                    "qcc.company_id," +
                    "qcc.company_name," +
                    "zs.INVITE_TITLE," +
                    "zs.INVITE_PUB_DATETIME," +
                    "zs.INVITE_NODENUM," +
                    "zs.BID_NOTICE_TYPE," +
                    "zs.jobid," +
                    "'1' as isadd " +
                    "from zs_tender zs left join " +
                    "tmp_t_eci_company qcc " +
                    "on upper(trim(qcc.credit_code)) = upper(trim(zs.creditcode)) where qcc.credit_code is not null");
            //删除
            spark.sql("insert into table a_dc_ep_incr.t_tender select " +
                    "qcc.id,\n" +
                    "qcc.key_no,\n" +
                    "qcc.company_id,\n" +
                    "qcc.company_name,\n" +
                    "qcc.title,\n" +
                    "qcc.public_date,\n" +
                    "qcc.province_name,\n" +
                    "qcc.channel_name,\n" +
                    "qcc.dates," +
                    "'-1' as isadd " +
                    "from qcc_tender qcc inner join " +
                    "(select a.* from a_dc_ep_ods.zs_t_invite_info_del a left join a_dc_ep_ods.zs_t_invite_info b " +
                    "on a.entid = b.entid and a.INVITE_PUB_DATETIME = b.INVITE_PUB_DATETIME where b.entid is null and a.creditcode is not null and a.creditcode !='') zs " +
                    "on upper(trim(qcc.credit_code)) = upper(trim(zs.creditcode)) and substr(trim(qcc.public_date),0,10) = trim(zs.INVITE_PUB_DATETIME)");
        }else {
            spark.sql("TRUNCATE TABLE a_dc_ep_incr.t_tender");
            System.out.println("中数本期数据涉及该表的数据为空。。。。。。");
        }
        spark.stop();
    }
}
